package com.haishuibutian.kafkaproducer.config;

import org.slf4j.Logger;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

@Component
public class KafkaClient {

    private static final Logger LOGGER = org.slf4j.LoggerFactory.getLogger(KafkaClient.class);

    private final KafkaTemplate<String,String> kafkaTemplate;

    public KafkaClient(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void send(String topic,String data){
        ListenableFuture<SendResult<String, String>> listenableFuture = kafkaTemplate.send(topic, data);
        listenableFuture.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onFailure(Throwable ex) {
                LOGGER.error("Kafka发送的消息失败 topic:{} data:{} exception:{}",topic,data,ex);
            }

            @Override
            public void onSuccess(SendResult<String, String> result) {
                LOGGER.info("Kafka发送消息成功 topic:{}",topic);
            }
        });
    }
}
